package com.bw.gmall.realtime.app.dwd;


import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//收藏商品事务事实表
/**
 * 互动域收藏事实表应用
 * 负责处理用户收藏商品的交互数据，将原始日志数据转化为结构化的收藏事实表
 * 主要功能点：
 * - 从Kafka读取基础日志数据
 * - 过滤出收藏相关的交互事件
 * - 提取收藏相关的维度信息（商品、用户、时间等）
 * - 将处理后的数据写入Kafka，供下游分析使用
 * 
 * 数据流向：
 * 基础日志数据 -> DwdInteractionFavorAdd -> 处理后写入dwd_interaction_favor_add主题
 */
public class DwdInteractionFavorAdd {
    /**
     * 主方法，处理收藏商品事务事实表数据
     * 1. 初始化Flink环境
     * 2. 从Kafka读取业务数据
     * 3. 过滤出收藏表数据
     * 4. 将处理后的数据写入Kafka目标表
     * @param args 命令行参数
     * @throws Exception 可能抛出的异常
     */
    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        // 创建流执行环境，设置并行度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2. 状态后端设置
        // 预留状态后端配置位置，可用于设置检查点和状态存储

        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        // 使用工具类创建Kafka源表topic_db
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("dwd_interaction_favor_add"));
        System.out.println("创建topic_db表成功");
        
        // 打印从topic_db表读取的数据，用于调试
        tableEnv.sqlQuery("select * from topic_db").execute().print();

        // TODO 4. 读取收藏表数据
        // 从topic_db中查询favor_info表的数据，筛选新增或未取消的收藏记录
        Table favorInfo = tableEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['sku_id'] sku_id, " +
                "date_format(data['create_time'],'yyyy-MM-dd') date_id, " +
                "data['create_time'] create_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'favor_info' " +
                "and (`type` = 'insert' or (`type` = 'update' and data['is_cancel'] = '0'))" +
                "");
        // 创建临时视图favor_info，便于后续SQL操作
        tableEnv.createTemporaryView("favor_info", favorInfo);

        // TODO 5. 创建 Kafka-Connector dwd_interaction_favor_add 表
        // 创建Kafka目标表，用于存储处理后的收藏数据
        tableEnv.executeSql("create table dwd_interaction_favor_add ( " +
                "id string, " +
                "user_id string, " +
                "sku_id string, " +
                "date_id string, " +
                "create_time string, " +
                "ts string " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_interaction_favor_add"));

        // TODO 6. 将数据写入 Kafka-Connector 表
        // 将临时视图favor_info中的数据插入到目标表dwd_interaction_favor_add
        tableEnv.executeSql("" +
                "insert into dwd_interaction_favor_add select * from favor_info").print();

        // TODO 7. 启动任务
        System.out.println("DwdInteractionFavorAdd任务启动中...");
        env.execute("DwdInteractionFavorAdd");
    }
}
